package kafka

import (
	"fmt"

	"github.com/micro-plat/hydra"
	"github.com/micro-plat/lib4go/types"
)

type KafkaService struct {
	Topic     string
	linesChan chan types.XMap
}

func NewKafkaService(topic string) *KafkaService {
	return &KafkaService{
		Topic:     topic,
		linesChan: make(chan types.XMap, 10000),
	}
}

func (k *KafkaService) RecvHandle(ctx hydra.IContext) interface{} {
	select {
	case k.linesChan <- ctx.Request().GetMap():
	default:
		return fmt.Errorf("队列已满，数据丢弃")
	}

	return "success"
}
func (k *KafkaService) Subscribe() <-chan types.XMap {
	return k.linesChan
}
